package edu.wisc.hevea.broker;

import org.apache.commons.io.FileUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import edu.wisc.cloudservice.CloudServiceClient;
import edu.wisc.hevea.constants.ApplicationConstants;
import edu.wisc.hevea.util.CommonUtil;

import java.io.File;
import java.io.IOException;
import java.util.Calendar;
import java.util.Date;
import java.util.HashMap;
import java.util.Map;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;

public class AutoScalingThread implements Runnable {

	private static final String QUEUE_METRIC_PREFIX = "/home/anand/Documents/DistributedSystems/HeveaLibraries/queue/";
	private static final String CONSUMER_SCALING_FILE = "/home/anand/Documents/DistributedSystems/HeveaLibraries/scaling/";

	
	
	private static final Logger logger = LoggerFactory
			.getLogger(AutoScalingThread.class);

	private MessageBroker messageBroker;
	private ConcurrentHashMap<String, Integer> dataCenterQueueLengthMap;
	private ConcurrentHashMap<String, Integer> brokerQueueLengthMap;
	private Map<String,Date> dataCenterScaledTime;
	
	private static Map<String, String> DATACENTER_TO_COLOUR_MAP;
		
	static {
		DATACENTER_TO_COLOUR_MAP = new HashMap<String, String>();
		DATACENTER_TO_COLOUR_MAP.put(ApplicationConstants.VIRGINIA_DC, "red");
		DATACENTER_TO_COLOUR_MAP.put(ApplicationConstants.IAD_DC, "green");
		DATACENTER_TO_COLOUR_MAP.put(ApplicationConstants.DUBLIN_DC, "gold");
		DATACENTER_TO_COLOUR_MAP.put(ApplicationConstants.HONGKONG_DC, "blue");
		
		
	}

	public AutoScalingThread(MessageBroker messageBroker) {
		this.messageBroker = messageBroker;
		brokerQueueLengthMap = new ConcurrentHashMap<String, Integer>();
		dataCenterQueueLengthMap = new ConcurrentHashMap<String, Integer>();
		dataCenterScaledTime = new HashMap<String,Date>();
		Calendar lastScaledTime = Calendar.getInstance();
		lastScaledTime.add(Calendar.SECOND, -60);
		dataCenterScaledTime.put(ApplicationConstants.VIRGINIA_DC, lastScaledTime.getTime());
		dataCenterScaledTime.put(ApplicationConstants.IAD_DC, lastScaledTime.getTime());
		dataCenterScaledTime.put(ApplicationConstants.DUBLIN_DC, lastScaledTime.getTime());
		dataCenterScaledTime.put(ApplicationConstants.HONGKONG_DC, lastScaledTime.getTime());
	}

	private long getSecondsForToday() {
		Calendar now = Calendar.getInstance();
		return now.get(Calendar.HOUR_OF_DAY) * 60 * 60
				+ now.get(Calendar.MINUTE) * 60 + now.get(Calendar.SECOND);
	}

	@Override
	public void run() {

		if (!messageBroker.isLeader()) {
			messageBroker.getMessagingServiceClient()
					.sendBrokerQueueLengthToLeader();
		} else {
			brokerQueueLengthMap.put(messageBroker.getMyHostName(),
					(int) messageBroker.getTopicManager()
							.getPendingQueueLength());
		}

		if (!messageBroker.isLeader()) {
			return;
		}

		dataCenterQueueLengthMap.clear();

		int virginiaDCQueueLength = 0;
		int iadDCQueueLength = 0;
		int dublinDCQueueLength = 0;
		int hongKongDCQueueLength = 0;

		short virginiaDCBrokerCount = 0;
		short iadDCBrokerCount = 0;
		short dublinDCBrokerCount = 0;
		short hongKongDCBrokerCount = 0;

		long secondsForToday = getSecondsForToday();
		Date currentTime = new Date();

		for (Entry<String, Integer> entry : brokerQueueLengthMap.entrySet()) {
			try {

				File graphFile = new File(QUEUE_METRIC_PREFIX + entry.getKey());

				if (!graphFile.exists()) {
					String color = DATACENTER_TO_COLOUR_MAP
							.get(ApplicationConstants.HOST_DC_MAPPING.get(entry
									.getKey()));
					FileUtils.write(graphFile,
							String.format("%s = %s\n", "color", color), false);
				}

				String queueLengthMetric = String.format("%d %d\n",
						secondsForToday, entry.getValue());
				FileUtils.write(graphFile, queueLengthMetric, true);

				if (ApplicationConstants.HOST_DC_MAPPING.get(entry.getKey())
						.equals(ApplicationConstants.VIRGINIA_DC)) {
					virginiaDCBrokerCount++;
					virginiaDCQueueLength += entry.getValue();
				} else if (ApplicationConstants.HOST_DC_MAPPING.get(
						entry.getKey()).equals(ApplicationConstants.IAD_DC)) {
					iadDCBrokerCount++;
					iadDCQueueLength += entry.getValue();
				} else if (ApplicationConstants.HOST_DC_MAPPING.get(
						entry.getKey()).equals(ApplicationConstants.DUBLIN_DC)) {
					dublinDCBrokerCount++;
					dublinDCQueueLength += entry.getValue();
				} else if (ApplicationConstants.HOST_DC_MAPPING.get(
						entry.getKey())
						.equals(ApplicationConstants.HONGKONG_DC)) {
					hongKongDCBrokerCount++;
					hongKongDCQueueLength += entry.getValue();
				}

			} catch (IOException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
		}

		dataCenterQueueLengthMap.put(ApplicationConstants.VIRGINIA_DC,
				virginiaDCQueueLength
						/ getQueueCountForAvg(virginiaDCBrokerCount));
		dataCenterQueueLengthMap.put(ApplicationConstants.IAD_DC,
				iadDCQueueLength / getQueueCountForAvg(iadDCBrokerCount));
		dataCenterQueueLengthMap.put(ApplicationConstants.DUBLIN_DC,
				dublinDCQueueLength / getQueueCountForAvg(dublinDCBrokerCount));
		dataCenterQueueLengthMap.put(ApplicationConstants.HONGKONG_DC,
				hongKongDCQueueLength
						/ getQueueCountForAvg(hongKongDCBrokerCount));

		for (String dataCenter : ApplicationConstants.AVAILABILITY_ZONES) {

			File graphFile = new File(QUEUE_METRIC_PREFIX + dataCenter);
			try {
				if (!graphFile.exists()) {
					String color = DATACENTER_TO_COLOUR_MAP.get(dataCenter);
					FileUtils.write(graphFile,
							String.format("%s = %s\n", "color", color), false);
				}
				String queueLengthMetric = String.format("%d %d\n",
						secondsForToday,
						dataCenterQueueLengthMap.get(dataCenter));
				FileUtils.write(graphFile, queueLengthMetric, true);
				
				
				
				if (dataCenterQueueLengthMap.get(dataCenter) > ApplicationConstants.LOW_WATER_MARK 
						&& CommonUtil.canInitiateScalingForConsumer(
								dataCenterScaledTime.get(dataCenter),
								currentTime)) {
					CloudServiceClient.getCloudServiceClient()
							.scaleConsumerInstances(dataCenter, 1);
					graphFile = new File(CONSUMER_SCALING_FILE + dataCenter);
					FileUtils.write(graphFile,
							String.format("%d %d\n",secondsForToday,1), true);
					dataCenterScaledTime.put(dataCenter, currentTime);
				}
				
			} catch (IOException ex) {
				ex.printStackTrace();
			}

		}
	}

	private int getQueueCountForAvg(int queueCount) {
		if (queueCount == 0) {
			return 1;
		} else {
			return queueCount;
		}
	}

	public ConcurrentHashMap<String, Integer> getDataCenterQueueLengthMap() {
		return dataCenterQueueLengthMap;
	}

	public void setDataCenterQueueLengthMap(
			ConcurrentHashMap<String, Integer> dataCenterQueueLengthMap) {
		this.dataCenterQueueLengthMap = dataCenterQueueLengthMap;
	}

	public ConcurrentHashMap<String, Integer> getBrokerQueueLengthMap() {
		return brokerQueueLengthMap;
	}

	public void setBrokerQueueLengthMap(
			ConcurrentHashMap<String, Integer> brokerQueueLengthMap) {
		this.brokerQueueLengthMap = brokerQueueLengthMap;
	}
}
